/* 
 * Enterprise Library Extensions - DI + ORM.
 * Copyright (C) 2006
 *
 * GNU General Public License
 *
 * This program is free software; you can redistribute 
 * it and/or modify it under the terms of the GNU General Public License 
 * as published by the Free Software Foundation; either version 2 of the 
 * License, or (at your option) any later version.
 * 
 * This program is distributed in the hope that it will be useful, but 
 * WITHOUT ANY WARRANTY; without even the implied warranty of 
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. 
 * See the GNU General Public License for more details.
 */
using System;
using System.Messaging;
using System.Reflection;
using ELE.Common.Util;
using ELE.EnterpriseLibrary.Common.Logging;
using ELE.EnterpriseLibrary.Naming;
using Microsoft.Practices.EnterpriseLibrary.Logging;

namespace ELE.EnterpriseLibrary.Messaging
{
  /// <summary>
  /// A queued message reader.
  /// 
  /// Author : Miguel Curi
  /// 
  /// </summary>
  public class MessageSink : LoggingObjectBase
  {
    private MessageQueue myQueue;
    private MessageSinkInfo myInfo;
    private IMessageProcessor myProcessor;
    /// <summary>
    /// Initializes a message sink that listens for messages on a queue defined by the given info
    /// </summary>
    /// <param name="info"></param>
    public MessageSink(MessageSinkInfo info) : this(info, false)
    {
    }
    /// <summary>
    /// Initializes a message sink that listens for messages on a queue defined by the given info 
    /// </summary>
    /// <param name="info">message queue info</param>
    /// <param name="logToContext">should we log errors to the static context ??</param>
    public MessageSink(MessageSinkInfo info, bool logToContext)
    {
      myInfo = info;
      Init();
    }
    /// <summary>
    /// Initializes a message sink that listens for messages on a queue defined by the given info 
    /// </summary>
    /// <param name="info">message queue info</param>
    /// <param name="context">The context that has the referenced processor</param>
    public MessageSink(MessageSinkInfo info, INamingContext context)
    {
      myInfo = info;
      Init(context);
    }
    /// <summary>
    /// Destructor closes the queue of this sink
    /// </summary>
    ~MessageSink()
    {
      if(myQueue != null)
      {
        myQueue.Close();
        myQueue = null;
      }
    }
    /// <summary>
    /// Close the underlaying queue
    /// </summary>
    public void Close()
    {
      if(myQueue != null)
      {
        myQueue.Close();
        myQueue = null;
      }
    }
    /// <summary>
    /// Initialization routine
    /// </summary>
    public void Init()
    {
      Init(null);
    }
    /// <summary>
    /// Initialization routine
    /// </summary>
    public void Init(INamingContext context)
    {
      if(myQueue != null)
      {
        myQueue.Close();
        myQueue = null;
      }
      if(myInfo.Processor != null)
      {
        try
        {
          object processor = null;
          if(context == null)
          {
            string className = myInfo.Processor.Substring(0, myInfo.Processor.IndexOf(",")).Trim();
            string assemblyName = myInfo.Processor.Substring(myInfo.Processor.IndexOf(",") + 1).Trim();
            Assembly assembly = Assembly.Load(assemblyName);
            processor = assembly.CreateInstance(className);
          }
          else
          {
            try
            {
              processor = context.Lookup(myInfo.Processor);
            }
            catch(ResourceInitializationException rie)
            {
              this.LogWarning("MessageSink: Error initializing message processor for message sink from context object named " + myInfo.Processor, rie);
            }
          }
          if(processor == null || !(typeof (IMessageProcessor)).IsAssignableFrom( processor.GetType()))
          {
            this.LogError("MessageSink: Processor associated with message queue named " + myInfo.QueueName + ", either the processor can't be instantiated or it is not an implementation of IMessageProcessor ");
            throw new NoProcessorDefinedException("MessageSink: Processor associated with message queue named " + myInfo.QueueName + ", either the processor can't be instantiated or it is not an implementation of IMessageProcessor ");

          }
          myProcessor = (IMessageProcessor) processor;
        }
        catch(Exception e)
        {        
          this.LogError("MessageSink: Had a problem initializing processor for message queue named " + myInfo.QueueName, e);
          throw new NoProcessorDefinedException("MessageSink: Had a problem initializing processor for message queue named " + myInfo.QueueName, e);
        }
        // Create an instance of MessageQueue. Set its formatter.
        try
        {
          //This only works for public queues
          if(myInfo.IsPublicQueue)
          {
            if(MessageQueue.Exists(myInfo.QueueName))
            {
              myQueue = new MessageQueue(myInfo.QueueName);
            }
            else
            {
              myQueue = MessageQueue.Create(myInfo.QueueName);
            }
          }
          else
          {
            myQueue = new MessageQueue(myInfo.QueueName);
          }
          myQueue.Formatter = new BinaryMessageFormatter();
          // Add an event handler for the ReceiveCompleted event.
          myQueue.ReceiveCompleted += 
            new ReceiveCompletedEventHandler(ReceiveCompleted);
          
          // Begin the asynchronous receive operation.
          myQueue.BeginReceive();
        }
        catch(System.InvalidOperationException ioe)
        {
          myQueue = null;
          this.LogError("MessageSink: The process failed to initialize the message queue at path : " + myInfo.QueueName, ioe);
          throw ioe;
        }
        catch(MessageQueueException mqe)
        {
          myQueue = null;
          this.LogError("MessageSink: The process failed to attach a listener to the message queue at path : " + myInfo.QueueName, mqe);
          throw mqe;
        }
      }
      else
      {
        this.LogError("MessageSink: Can not initialize message sink without a processor, make sure you defined one in the config file for " + myInfo.QueueName);
        throw new NoProcessorDefinedException("MessageSink: Can not initialize message sink without a processor, make sure you defined one in the config filefor " + myInfo.QueueName);
      }
    }
    /// <summary>
    /// Exposes the message queue
    /// </summary>
    public MessageQueue Queue
    {
      get { return myQueue; }
    }
    /// <summary>
    /// Exposes the sink info
    /// </summary>
    public MessageSinkInfo SinkInfo
    {
      get { return myInfo; }
    }
    /// <summary>    
    /// Provides an event handler for the ReceiveCompleted
    /// event.
    /// </summary>
    /// <param name="source"></param>
    /// <param name="asyncResult"></param>
    
    private void ReceiveCompleted(Object source, ReceiveCompletedEventArgs asyncResult)
    {
      try
      {
        // Connect to the queue.
        MessageQueue mq = (MessageQueue)source;
        
        // End the asynchronous receive operation.
        Message m = mq.EndReceive(asyncResult.AsyncResult);
        
        if(myProcessor != null)
        {
          myProcessor.Process(m);
        }
        
        // Restart the asynchronous receive operation.
        mq.BeginReceive();
      }
      catch(MessageQueueException mqe)
      {
        this.LogError("MessageSink: Encountered a Messaging error processing message from queue " + myInfo.QueueName, mqe);
      }
      catch(Exception e)
      {
        this.LogError("MessageSink: Encountered an error processing message from queue " + myInfo.QueueName, e);
      }
    }

  }
}
